package net.bwie.realtime.jtp.dim.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.dim.function.DimDataFilterFunction;
import net.bwie.realtime.jtp.dim.function.HbaseDimMapFunction;
import net.bwie.realtime.jtp.dim.sink.DimHbaseSinkFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;

/**
 * @Author: FuHe
 * @Date: 2025/6/2
 */
public class RealtimeSyncDataDimJob {
    public static void main(String[] args) throws Exception {
        // 1 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //  2 读取kafka数据源
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "topic-db");
//        kafkaStream.print();
        // 3 数据转换-transformation
        DataStream<String> dimStream = handle(kafkaStream);
//        dimStream.print();
        // 4 数据接收器-sink
        dimStream.addSink(new DimHbaseSinkFunction());
        // 5 触发执行-execute
        env.execute("RealtimeSyncDataDimJob");
    }

    private static DataStream<String> handle(DataStream<String> kafkaStream) {
        // 维度表名称与主键映射
        HashMap<String, String> dimMap = new HashMap<String, String>() {
            {
                put("base_dic", "dic_code");//
                put("base_province", "id");//
                put("base_region", "id");//
                put("base_category1", "id");//
                put("base_category2", "id");//
                put("base_category3", "id");//
                put("sku_info", "id");//
                put("spu_info", "id");//
                put("user_info", "id");//
                put("activity_info", "id");
                put("activity_rule", "id");//
                put("coupon_info", "id");//
                put("base_trademark", "id");//
            }
        };
        // 维度表名称
        HashSet<String> dimSet = new HashSet<>(dimMap.keySet());
        // 1 过滤维表数据
        SingleOutputStreamOperator<String> syncStream = kafkaStream.filter(new DimDataFilterFunction(dimSet));
//        syncStream.print();
        SingleOutputStreamOperator<String> hbaseStream = syncStream.map(new HbaseDimMapFunction(dimMap));
//        hbaseStream.print();
        return hbaseStream;
    }
}
